
/*
 * Copyright © 2021 https://www.cestc.cn/ All rights reserved.
 */

package com.zx.learn.flink.state;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 使用KeyState中的ValueState获取数据中的最大值(实际中直接使用maxBy即可)，使用值状态自定义.
 */
public class KeyStateDemo {
    public static void main(String[] args) throws Exception {
        //1.env 设置并发度为1
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //2.Source 参看课件
        DataStreamSource<Tuple2<String, Long>> tupleDS = env.fromElements(
                Tuple2.of("北京", 1L),
                Tuple2.of("上海", 2L),
                Tuple2.of("北京", 6L),
                Tuple2.of("上海", 8L),
                Tuple2.of("北京", 3L),
                Tuple2.of("上海", 4L)
        );
        //3.Transformation
        //使用KeyState中的ValueState获取流数据中的最大值(实际中直接使用maxBy即可)
        //实现方式1:直接使用maxBy--开发中使用该方式即可
        DataStream<Tuple2<String, Long>> result1 = tupleDS.keyBy(t -> t.f0)
                .maxBy(1);

        //min只会求出最小的那个字段,其他的字段不管
        //minBy会求出最小的那个字段和对应的其他的字段
        //max只会求出最大的那个字段,其他的字段不管
        //maxBy会求出最大的那个字段和对应的其他的字段
        //实现方式2:通过managed state输入的state
        //Tuple2<String/*单词*/, Long/*长度*/> 输出 Tuple3<String/*单词*/, Long/*长度*/, Long/*历史最大值*/>
        SingleOutputStreamOperator<Tuple3<String, Long, Long>> maxCount = tupleDS.keyBy(t -> t.f0)
                .map(new RichMapFunction<Tuple2<String, Long>, Tuple3<String, Long, Long>>() {
                    //保存当前内存中最大的值的state
                    private transient ValueState<Long> currentMaxValue;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //存储到内存中的数据结构的描述
                        ValueStateDescriptor desc = new ValueStateDescriptor("maxCount", TypeInformation.of(Long.class));
                        currentMaxValue = getRuntimeContext().getState(desc);
                    }

                    //3.3.重写 map 方法
                    //-4.获取state中历史最大值value和当前元素的最大值并比较
                    @Override
                    public Tuple3<String, Long, Long> map(Tuple2<String, Long> value) throws Exception {
                        String city = value.f0;
                        Long currentValue = value.f1;
                        if (currentMaxValue.value() == null || currentMaxValue.value() < currentValue) {
                            currentMaxValue.update(currentValue);
                            return Tuple3.of(city, currentValue, currentMaxValue.value());
                        } else {
                            return Tuple3.of(city, currentValue, currentMaxValue.value());
                        }
                    }
                });
        //3.1.先根据字符串f0分组然后进行 map 操作，将Tuple2<String/*单词*/, Long/*长度*/> 输出 Tuple3<String/*单词*/, Long/*长度*/, Long/*历史最大值*/>
        //-1.定义值类型的状态用来存储最大值
        //3.2.重写 RichMapFunction 的open 方法
        //-2.定义状态描述符
        //-3.从当前上下文获取内存中的状态值
        //3.3.重写 map 方法
        //-4.获取state中历史最大值value和当前元素的最大值并比较
        //-5.如果当前值大或历史值为空更新状态；返回Tuple3元祖结果
        //4.Sink 打印输出
//        result1.printToErr();
        maxCount.print();
        //5.execute 执行环境
        env.execute();
    }
}
